Amplifyで構築したAppSyncへPythonで接続する

kobayashi.m

2020.09.04

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

データアナリティクス事業本部のkobayashiです。

最近の業務でAmplifyとReact使った開発を行っており、その中ではデータベースとしてDynamoDBをAppSync経由で使用しています。その際にフロントエンドはAmplifyを使えば簡単にデータベースと読み書きを簡単に行えるのですが、バックエンド用途として重たい処理をPythonを使って別の仕組みで実装し、その仕組の中でデータベースの読み書きを行いたい場合に色々方法を検討しました。その中で今回採用した方法をまとめたいと思います。

環境

Python 3.7.4
requests-aws4auth 1.0
gql 2.0.0

AppSyncをPythonで扱う

前提条件

AmplifyとReactでフロントを構築しているため、GraphQLのスキーマはschema.graphqlにて定義し、amplify pushしてAppSync側に適用しています。その際にフロントはCognito認証にてAppSyncへ接続を行いますが、バックエンド用途のPythonはCognitoで発行されたアクセストークンとIAMの2パターンで試してみるためAmplify公式の情報を参考に下記の様に設定しました。

公式ドキュメント

API (GraphQL) - Directives - Amplify Docs

type UserProfile
  @model
  @auth(rules: [
    { allow: owner },
    { allow: private, provider: iam }
  ])
  @key(fields: ["username"])
{
  username: ID!
  city: String
}

ポイントはauthディレクティブに認証プロバイダとしてIAMを追加している点です。この記述によりCongitoのIDプールで認証された権限としてIAM認証を使ってAppSyncにアクセスできるようになります。

Pythonモジュールgplのインストール

gplモジュールはPython用のGraphQLクライアントになります。このモジュールを使うことで簡単にGraphQLを扱うことができるので今回採用しました。

インストールは簡単でいつも通りpipを使います。

pip install gql

また併せてIAMの場合はV4署名リクエストを行う必要があるのでrequests-aws4authをインストールします。

pip install requests-aws4auth

requests-aws4authの使い方については弊社ブログで検証エントリがありますのでそちらをご確認ください。

以上で必要なモジュールがインストールできました。

ここから先の流れとしては

認証方法に従ってGraphQLクライントを作成
Query,Mutationを行う

となります。

アクセストークンを使ったパターン

アクセストークンでGraphQLクライアントを作成する部分は以下の様になります。

from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport

headers = {
    "Accept": "application/json",
    "Content-Type": "application/json",
    "Authorization": "アクセストークン"
}

transport = RequestsHTTPTransport(
    url="AppSyncエンドポイント",
    use_json=True,
    headers=headers
)
client = Client(transport=transport,
                fetch_schema_from_transport=True)

ヘッダーにAuthorizationとしてアクセストークンを与えれば良いだけです。

IAMを使ったパターン

IAMでGraphQLクライアントを作成する部分は以下の様になります。

from requests_aws4auth import AWS4Auth
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport

auth = AWS4Auth(
    "アクセスキー",
    "シークレットキー",
    "ap-northeast-1",
    "appsync",
)
headers = {
    "Accept": "application/json",
    "Content-Type": "application/json",
}

transport = RequestsHTTPTransport(
    url="AppSyncエンドポイント",
    use_json=True,
    headers=headers,
    auth=auth
)
client = Client(transport=transport,
                fetch_schema_from_transport=True)

アクセストークンを使った場合と3箇所異なります。

requests_aws4authでV4署名リクエストを行う
headerからAuthorizationを削除する
GraphQLクライアント作成時にauthパラメータを追加する

Queryのテスト

それでは実際にQueryを試してみます。テストなのでgetとlistの2パターンを試してみたいと思います。コードは以下のような形になります。

def query_get():
    client = get_client()
    query = """
            query getUserProfile($username: ID!) {
                getUserProfile(username:$username){
                    username
                    city
                    createdAt
                    updatedAt
                }
            }
        """
    params = {"username": "test1"}
    resp = client.execute(gql(query),
                          variable_values=json.dumps(params))
    pprint.pprint(resp)


def query_list():
    client = get_client()
    query = """
            query listUserProfiles {
                listUserProfiles{
                    items {
                        username
                        city
                        createdAt
                        updatedAt
                    }
                }
            }
        """
    resp = client.execute(gql(query))
    pprint.pprint(resp)

実行してみると、以下の様に指定した属性が取得できています。

・query_get()の実行結果

{'getUserProfile': {'createdAt': '2020-09-01T22:35:21.069Z',
                  'city': 'tokyo',
                  'updatedAt': '2020-09-01T22:35:21.069Z',
                  'username': 'test1'}}

・query_list()の実行結果

{'listUserProfiles': {'items': [{'createdAt': '2020-09-01T22:35:58.318Z',
                               'city': 'osaka',
                               'updatedAt': '2020-09-01T22:35:58.318Z',
                               'username': 'test3'},
                              {'createdAt': '2020-09-01T22:35:24.846Z',
                               'city': 'tokyo',
                               'updatedAt': '2020-09-01T22:35:24.846Z',
                               'username': 'test2'},
                              {'createdAt': '2020-09-01T22:36:29.816Z',
                               'city': 'nagoya',
                               'updatedAt': '2020-09-01T22:36:29.816Z',
                               'username': 'test4'},
                              {'createdAt': '2020-09-01T22:37:24.306Z',
                               'city': 'yokohama',
                               'updatedAt': '2020-09-01T22:37:24.306Z',
                               'username': 'test5'},
                              {'createdAt': '2020-09-01T22:35:21.069Z',
                               'city': 'tokyo',
                               'updatedAt': '2020-09-01T22:35:21.069Z',
                               'username': 'test1'}]}}

Mutationのテスト

次にMutasionのテストをしてみます。Mutationのコードは以下になります。

def mutation_create():
    client = get_client()
    params = {"username": "test6", "token": "sapporo"}
    query = """
        mutation createUserProfile($input: CreateUserProfileInput!) {
            createUserProfile(input: $input){
                username
                city
            }
        }
    """
    resp = client.execute(gql(query),
                          variable_values=json.dumps({"input": params}))
    pprint.pprint(resp)


def mutation_update():
    client = get_client()
    params = {"username": "test6", "city": "fukuoka"}
    query = """
        mutation updateUserProfile($input: UpdateUserProfileInput!) {
            updateUserProfile(input: $input){
                username
                city
            }
        }
    """
    resp = client.execute(gql(query),
                          variable_values=json.dumps({"input": params}))
    pprint.pprint(resp)


def mutation_delete():
    client = get_client()
    params = {"username": "test1"}
    query = """
        mutation deleteUserProfile($input: DeleteUserProfileInput!) {
            deleteUserProfile(input: $input){
                username
                city
            }
        }
    """
    resp = client.execute(gql(query),
                          variable_values=json.dumps({"input": params}))
    pprint.pprint(resp)

実行してみると、こちらも以下の様に指定した属性が取得できています。

・mutation_create()の実行結果

{'createUserProfile': {'city': 'sapporo', 'username': 'test6'}}

・mutation_update()の実行結果

{'updateUserProfile': {'city': 'fukuoka', 'username': 'test6'}}

・mutation_delete()の実行結果

{'deleteUserProfile': {'city': 'fukuoka', 'username': 'test6'}}

まとめ

Amplifyで構築したAppSyncへPythonから「アクセストークン」、「IAM」を使って接続してみました。時間のかかる処理をAmplifyとは別に実装するがAppSyncは扱いたいという場面はそこそこあるのではないかと思いますのでご参考になりましたら幸いです。

最後まで読んで頂いてありがとうございました。

Amplifyで構築したAppSyncへPythonで接続する

はじめに

環境

AppSyncをPythonで扱う

前提条件

Pythonモジュールgplのインストール

アクセストークンを使ったパターン

IAMを使ったパターン

Queryのテスト

Mutationのテスト

まとめ

関連記事

主なカテゴリ

AWS

おすすめ

プロダクト

コンテンツ

お問い合わせ

運営会社